#!/bin/bash

if [ -n "$1" ] ; then
  data_date=$1
else
  data_date=`date -d '-1 days' +%F`
fi

TMP_DWD_ADS_EVENT_LOG_PARSE_MYSQL="
DROP TABLE IF EXISTS jtp_warehouse.tmp_dwd_ads_event_log_parse;
CREATE TABLE IF NOT EXISTS jtp_warehouse.tmp_dwd_ads_event_log_parse
AS
SELECT
    parse_url('hdfs://www.baidu.com' || request_url ,'QUERY','t') AS event_time
    ,split(parse_url('hdfs://www.baidu.com' || request_url ,'PATH'),'/')[3] AS event_type
    ,parse_url('hdfs://www.baidu.com' || request_url ,'QUERY','id') AS ads_id
    ,split(parse_url('hdfs://www.baidu.com' || request_url ,'PATH'),'/')[2] AS platform_name_en
    ,parse_url('hdfs://www.baidu.com' || request_url ,'QUERY','ip') AS client_ip
    ,parse_url('hdfs://www.baidu.com' || request_url ,'QUERY','device_id') AS client_device_id
    ,parse_url('hdfs://www.baidu.com' || request_url ,'QUERY','os_type') AS client_os_type
    ,reflect('java.net.URLDecoder','decode',parse_url('hdfs://www.baidu.com' || request_url ,'QUERY','ua'),'utf-8') AS client_user
FROM jtp_warehouse.ods_ads_log_inc
WHERE dt = '${data_date}'
;
"

TMP_DWD_ADS_EVENT_LOG_DIM_MYSQL="
USE jtp_warehouse;
DROP TABLE IF EXISTS tmp_dwd_ads_event_log_dim;
CREATE TABLE IF NOT EXISTS tmp_dwd_ads_event_log_dim
AS
SELECT
    t1.event_time
     ,t1.event_type
     ,t1.ads_id
     ,t2.ad_name
     ,t2.product_id AS ads_product_id
     ,t2.product_name AS ads_product_name
     ,t2.product_price AS ads_product_price
     ,t2.material_id AS ads_material_id
     ,t2.ad_group_id AS ads_group_id
     ,t2.platform_id
     ,t2.platform_name AS platform_name_en
     ,t2.platform_name_zh
     ,t1.client_ip
     ,t1.client_device_id
     ,t1.client_os_type
     ,t1.client_user AS client_user_agent
FROM jtp_warehouse.tmp_dwd_ads_event_log_parse t1
         LEFT JOIN (
    SELECT
        ad_id, ad_name, ad_group_id, product_id
         , product_name, product_price, material_id
         , platform_id, platform_name, platform_name_zh
    FROM jtp_warehouse.dim_ads_platform_info_full
    WHERE dt = '${data_date}'
) t2 ON t1.ads_id = t2.ad_id
;
"


TMP_DWD_ADS_EVENT_LOG_REGION_MYSQL="
DROP TABLE IF EXISTS jtp_warehouse.tmp_dwd_ads_event_log_region;
CREATE TABLE IF NOT EXISTS jtp_warehouse.tmp_dwd_ads_event_log_region
AS
SELECT
    event_time
     , event_type
     , ads_id, ad_name
     , ads_product_id
     , ads_product_name
     , ads_product_price
     , ads_material_id
     , ads_group_id
     , platform_id
     , platform_name_en
     , platform_name_zh
     ,region_map['country'] AS client_country
     ,region_map['area'] AS client_area
     ,region_map['province'] AS client_province
     ,region_map['city'] AS client_city
     , client_ip
     , client_device_id
     , client_os_type
     , client_user_agent
FROM(
        SELECT
            *,jtp_ads_warehouse.ip_to_region(client_ip) AS region_map
        FROM jtp_warehouse.tmp_dwd_ads_event_log_dim
    )t1
;
"


TMP_DWD_ADS_EVENT_LOG_BROWSER_MYSQL="
DROP TABLE IF EXISTS jtp_warehouse.tmp_dwd_ads_event_log_browser;
CREATE TABLE IF NOT EXISTS jtp_warehouse.tmp_dwd_ads_event_log_browser
AS
SELECT
    event_time
     , event_type
     , ads_id, ad_name
     , ads_product_id
     , ads_product_name
     , ads_product_price
     , ads_material_id
     , ads_group_id
     , platform_id
     , platform_name_en
     , platform_name_zh
     , client_country
     , client_area
     , client_province
     , client_city
     , client_ip
     , client_device_id
     , client_os_type
     ,browser_map['os_version'] AS client_os_version
     ,browser_map['browser'] AS client_browser_type
     ,browser_map['browser_version'] AS client_browser_version
     , client_user_agent
FROM(
        SELECT
            *,jtp_ads_warehouse.ua_to_browser(client_user_agent) AS browser_map
        FROM jtp_warehouse.tmp_dwd_ads_event_log_region
    )t1
;
"


TMP_DWD_ADS_EVENT_LOG_TRAFFIC_MYSQL="
DROP TABLE IF EXISTS tmp_dwd_ads_event_log_traffic;
CREATE TABLE IF NOT EXISTS tmp_dwd_ads_event_log_traffic
AS
WITH tmp_ip AS (
    SELECT
        DISTINCT client_ip
    FROM(
            SELECT
                client_ip,ads_id,event_time
                 ,count(1) OVER (
                PARTITION BY client_ip,ads_id
                ORDER BY CAST(event_time AS BIGINT)
                RANGE BETWEEN 300000 PRECEDING AND CURRENT ROW
                ) AS cnt
            FROM jtp_warehouse.tmp_dwd_ads_event_log_parse
        )t1
    WHERE cnt > 100

    UNION

    SELECT
        DISTINCT client_ip
    FROM(
            SELECT
                client_ip,ads_id,event_time,next_time
                 ,(next_time - event_time) AS interval_me
            FROM(
                    SELECT
                        client_ip,ads_id,event_time
                         ,lead(event_time,1,0) over (PARTITION BY client_ip,ads_id ORDER BY event_time) AS next_time
                    FROM tmp_dwd_ads_event_log_parse
                )t1
        )t2
    GROUP BY client_ip,ads_id,interval_me
    HAVING count(1) > 5
),tmp_device_id AS (
    SELECT
        DISTINCT client_device_id
    FROM(
            SELECT
                client_device_id,ads_id,event_time
                 ,count(1) OVER (
                PARTITION BY client_device_id,ads_id
                ORDER BY CAST(event_time AS BIGINT)
                RANGE BETWEEN 300000 PRECEDING AND CURRENT ROW
                ) AS cnt
            FROM jtp_warehouse.tmp_dwd_ads_event_log_parse
        )t1
    WHERE cnt > 100

    UNION

    SELECT
        DISTINCT client_device_id
    FROM(
            SELECT
                client_device_id, ads_id, event_time, next_time
                 ,(next_time - event_time) AS interval_me
            FROM(
                    SELECT
                        client_device_id,ads_id,event_time
                         ,lead(event_time,1,0) OVER (PARTITION BY client_device_id,ads_id ORDER BY event_time) AS next_time
                    FROM jtp_warehouse.tmp_dwd_ads_event_log_parse
                )t1
        )t2
    GROUP BY client_device_id,ads_id,interval_me
    HAVING count(1) > 5
)
SELECT
    t1.event_time
    ,t1.event_type
    ,t1.ads_id
    ,t1.ad_name AS ads_name
    ,t1.ads_product_id
    ,t1.ads_product_name
    ,t1.ads_product_price
    ,t1.ads_material_id
    ,t1.ads_group_id
    ,t1.platform_id
    ,t1.platform_name_en
    ,t1.platform_name_zh
    ,t1.client_country
    ,t1.client_area
    ,t1.client_province
    ,t1.client_city
    ,t1.client_ip
    ,t1.client_device_id
    ,t1.client_os_type
    ,t1.client_os_version
    ,t1.client_browser_type
    ,t1.client_browser_version
    ,t1.client_user_agent
    ,t2.client_ip IS NULL AND t3.client_device_id IS NULL AS is_invalid_traffic
FROM jtp_warehouse.tmp_dwd_ads_event_log_browser t1
    LEFT JOIN tmp_ip t2 ON t1.client_ip = t2.client_ip
    LEFT JOIN tmp_device_id t3 ON t1.client_device_id = t3.client_device_id
;
"


DWD_ADS_EVENT_LOG_LNC_MYSQL="
INSERT OVERWRITE TABLE jtp_warehouse.dwd_ads_event_log_lnc PARTITION (dt = '${data_date}')
SELECT
    CAST(event_time AS BIGINT)
     ,event_type
     ,ads_id
     ,ads_name
     ,ads_product_id
     ,ads_product_name
     ,ads_product_price
     ,ads_material_id
     ,ads_group_id
     ,platform_id
     ,platform_name_en
     ,platform_name_zh
     ,client_country
     ,client_area
     ,client_province
     ,client_city
     ,client_ip
     ,client_device_id
     ,client_os_type
     ,client_os_version
     ,client_browser_type
     ,client_browser_version
     ,client_user_agent
     ,is_invalid_traffic
FROM jtp_warehouse.tmp_dwd_ads_event_log_traffic
;
"

/opt/module/spark/bin/beeline -u jdbc:hive2://node101:10001 -n bwie -e  \
"${TMP_DWD_ADS_EVENT_LOG_PARSE_MYSQL}${TMP_DWD_ADS_EVENT_LOG_DIM_MYSQL}${TMP_DWD_ADS_EVENT_LOG_REGION_MYSQL}
${TMP_DWD_ADS_EVENT_LOG_BROWSER_MYSQL}${TMP_DWD_ADS_EVENT_LOG_TRAFFIC_MYSQL}${DWD_ADS_EVENT_LOG_LNC_MYSQL}"